package com.zimin.cloud.user.websocket;

import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSONObject;
import com.zimin.cloud.user.common.constant.UserMomentsConstant;
import com.zimin.common.utils.StringUtils;
import com.zimin.redis.utils.RedisOperator;
import com.zimin.rocketmq.utils.RocketMQUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 恒利
 * @since 2022/7/5 9:42 PM
 * WebSocketService是多例的，所以在注入Bean时需要特殊处理
 */
@Slf4j
@Component
@ServerEndpoint("/im-serer")
public class WebSocketService {

    /**
     * 在线数量
     */
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    /**
     * 存放所有的客户端的service
     */
    public static final ConcurrentHashMap<String, WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    /**
     * 服务端与客户端的会话
     */
    private Session session;

    /**
     * 唯一标识
     */
    private String sessionId;

    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }

    /**
     * 全局 redis参数
     */
    private static RedisOperator REDIS_OPERATOR = SpringUtil.getBean(RedisOperator.class);
    private static DefaultMQProducer PRODUCER = SpringUtil.getBean("danmusProducer");

    /**
     * <p>开始创建连接</p>
     */
    @OnOpen
    private void openConnection(Session session) {
        this.sessionId = session.getId();
        this.session = session;
        if (WEBSOCKET_MAP.containsKey(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId, this);
        } else {
            WEBSOCKET_MAP.put(sessionId, this);
            ONLINE_COUNT.getAndIncrement();
        }
        log.info("用户连接成功:{},当前在线人数为:{}", sessionId, ONLINE_COUNT);
        try {
            this.sendMessage("0");
        } catch (Exception e) {
            log.error("连接异常");
        }
    }

    @OnClose
    public void closeConnection() {
        if (WEBSOCKET_MAP.containsKey(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement();
        }
        log.info("用户退出:{},当前在线人数为:{}", sessionId, ONLINE_COUNT);
    }

    @OnMessage
    public void onMessage(String message) {
        log.info("用户信息:{}，报文:{}", sessionId, message);
        if (StringUtils.isNotEmpty(message)) {
            try {
                //群发消息
                WEBSOCKET_MAP.forEach((k, v) -> {
                    //把消息放在rocketmq中,当队列有时间时再发送消息
                    var jsonObject = new JSONObject();
                    jsonObject.put("message", message);
                    jsonObject.put("sessonId", v.getSessionId());
                    var msg = new Message(UserMomentsConstant.TOPIC_DAN_MU, JSONObject.toJSONString(jsonObject).getBytes());
                    try {
                        RocketMQUtil.asyncSendMsg(PRODUCER, msg);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
                //弹幕信息保存到数据库
            } catch (Exception e) {
                log.error("弹幕接收出现问题");
                e.printStackTrace();
            }
        }
    }

    @OnError
    public void onError(Throwable error) {

    }

    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }
}
